package com.example.demo.kafka;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

//每隔五秒，将过去是10秒内，通话时间最长的通话日志输出。
public class WaterMarkDemo {
	public static void main(String[] args) throws Exception {
		//得到Flink流式处理的运行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		env.setParallelism(1);
		//设置周期性的产生水位线的时间间隔。当数据流很大的时候，如果每个事件都产生水位线，会影响性能。
		env.getConfig().setAutoWatermarkInterval(100);//默认100毫秒


		ArrayList<StationLog> list = new ArrayList<>();

		list.add(new StationLog("a","b","1",1,1000));
		list.add(new StationLog("b","c","1",1,1));
		list.add(new StationLog("a","b","1",1,1));
		list.add(new StationLog("a","b","1",1,1001));
		//得到输入流
		DataStreamSource<StationLog> stream = env.fromCollection(list);
		stream.filter(new FilterFunction<StationLog>() {
			
			@Override
			public boolean filter(StationLog value) throws Exception {
				return value.getDuration() > 0?true:false;
			}
		}).assignTimestampsAndWatermarks(WatermarkStrategy.<StationLog>forBoundedOutOfOrderness(Duration.ofSeconds(3))
				.withTimestampAssigner(new SerializableTimestampAssigner<StationLog>() {
					@Override
					public long extractTimestamp(StationLog element, long recordTimestamp) {
						return element.getCallTime(); //指定EventTime对应的字段
					}
				})
		).keyBy(new KeySelector<StationLog, String>(){
			@Override
			public String getKey(StationLog value) throws Exception {
				return value.getStationID();  //按照基站分组
			}}
		).timeWindow(Time.seconds(1000)) //设置时间窗口
		.process(new MyProcessWindows())
				.print();
//		.reduce(new MyReduceFunction(),new MyProcessWindows()).print();

		env.execute();
	}
}
//用于如何处理窗口中的数据，即：找到窗口内通话时间最长的记录。
class MyReduceFunction implements ReduceFunction<StationLog> {
	@Override
	public StationLog reduce(StationLog value1, StationLog value2) throws Exception {
		// 找到通话时间最长的通话记录
		return value1.getDuration() >= value2.getDuration() ? value1 : value2;
	}
}
//窗口处理完成后，输出的结果是什么
class MyProcessWindows extends ProcessWindowFunction<StationLog, String, String, TimeWindow> {
	@Override
	public void process(String key, ProcessWindowFunction<StationLog, String, String, TimeWindow>.Context context,
			Iterable<StationLog> elements, Collector<String> out) throws Exception {
//		StationLog maxLog = elements.iterator().next();
		for (StationLog maxLog : elements) {
			StringBuffer sb = new StringBuffer();
			sb.append("窗口范围是:").append(context.window().getStart()).append("----").append(context.window().getEnd()).append("\n");;
			sb.append("基站ID：").append(maxLog.getStationID()).append("\t")
					.append("呼叫时间：").append(maxLog.getCallTime()).append("\t")
					.append("主叫号码：").append(maxLog.getFrom()).append("\t")
					.append("被叫号码：")	.append(maxLog.getTo()).append("\t")
					.append("通话时长：").append(maxLog.getDuration()).append("\n");
			out.collect(sb.toString());
		}


	}
}
